package com.jxw.cloudpen.web.netty.websocketclient;

import com.alibaba.fastjson.JSONObject;
import com.google.api.gax.rpc.ResponseObserver;
import com.google.cloud.speech.v1p1beta1.StreamingRecognizeResponse;
import com.jxw.cloudpen.web.netty.Entity.StreamData;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;

import javax.sound.sampled.*;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

public class MyWebSocketClientV2 extends WebSocketClient {
    ExecutorService executorService = Executors.newFixedThreadPool(3);
    public MyWebSocketClientV2(URI serverUri) {
        super(serverUri);
    }

    @Override
    public void onOpen(ServerHandshake handshakedata) {
        System.out.println("Connected");
//        this.send("Hello Server!"); // 发送消息到服务器
//        for(;;){
//            this.send(ByteObjUtil.objToBytes("aaaa"));
//            this.send("aaaa");
//            try {
//                Thread.sleep(3000);
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//        }
//        StreamData streamData = new StreamData();
//        streamData.setType(1);
//        streamData.setClientId("f544f5209e7d4f59a82ba3a780736faf");
//        this.send(ByteObjUtil.objToBytes(streamData));

        StreamData result = new StreamData();
        result.setType(1);
        result.setLanguageCode("zh-CN");
//        this.send(ByteObjUtil.objToBytes(result));
        this.send(JSONObject.toJSONString(result));
//        try {
//            Thread.sleep(3000);
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }

    }


    private static TargetDataLine targetDataLine;

    private static volatile BlockingQueue<byte[]> sharedQueue = new LinkedBlockingQueue<byte[]>();


    public StreamData executeMusic(String clientId) throws LineUnavailableException {





        // SampleRate:16000Hz, SampleSizeInBits: 16, Number of channels: 1, Signed: true,
        // bigEndian: false
        AudioFormat audioFormat = new AudioFormat(16000, 16, 1, true, false);
        DataLine.Info targetInfo =
                new DataLine.Info(
                        TargetDataLine.class,
                        audioFormat); // Set the system information to read from the microphone audio
        // stream

        if (!AudioSystem.isLineSupported(targetInfo)) {
            System.out.println("Microphone not supported");
            System.exit(0);
        }
        // Target data line captures the audio stream the microphone produces.
        targetDataLine = (TargetDataLine) AudioSystem.getLine(targetInfo);
        targetDataLine.open(audioFormat);

//        StreamData result = new StreamData();
// Microphone Input buffering
        class MicBuffer implements Runnable {

            @Override
            public void run() {
                System.out.println("Start speaking...Press Ctrl-C to stop");
                targetDataLine.start();
                byte[] data = new byte[6400];
                while (targetDataLine.isOpen()) {
                    try {
                        int numBytesRead = targetDataLine.read(data, 0, data.length);
                        if ((numBytesRead <= 0) && (targetDataLine.isOpen())) {
                            continue;
                        }
                        //组装数据
                        sharedQueue.put(data.clone());
                    } catch (InterruptedException e) {
                        System.out.println("Microphone input buffering interrupted : " + e.getMessage());
                    }
                }
            }
        }

        // Creating microphone input buffer thread
        MicBuffer micrunnable = new MicBuffer();
        Thread micThread = new Thread(micrunnable);
        micThread.start();
        ResponseObserver<StreamingRecognizeResponse> responseObserver = null;

        executorService.submit(()-> {
            try {
                if (!AudioSystem.isLineSupported(targetInfo)) {
                    System.out.println("Microphone not supported");
                    System.exit(0);
                }
                long startTime = System.currentTimeMillis();
                while (true) {
                    byte[] bytes=  sharedQueue.take();
                    if(bytes!=null && bytes.length>0){
                        StreamData result = new StreamData();
                        result = new StreamData();
                        result.setClientId(clientId);
                        result.setData(bytes);
                        result.setType(2);
//                bootstrap.channel.writeAndFlush(result);
//                    bootstrap.channel.writeAndFlush(ByteObjUtil.objToBytes(result));
//                        this.send(ByteObjUtil.objToBytes(result));

                        this.send(JSONObject.toJSONString(result));
                    }

                }
            } catch (Exception e) {
                System.out.println(e);
            }
        });
        return null;
    }


    @Override
    public void onMessage(String message) {
        System.out.println("Received: " + message);

        JSONObject jsonObject =JSONObject.parseObject(message);
        jsonObject.getString("clientId");
        try {
            executeMusic(jsonObject.getString("clientId"));
        } catch (LineUnavailableException e) {
            e.printStackTrace();
        }


    }

    @Override
    public void onMessage(ByteBuffer bytes) {
        System.out.println("Received: " + bytes);
    }

    @Override
    public void onClose(int code, String reason, boolean remote) {
        System.out.println("Disconnected");
    }

    @Override
    public void onError(Exception ex) {
        System.err.println("Error: " + ex.getMessage());
    }

    public static void main(String[] args) throws URISyntaxException {
        MyWebSocketClientV2 client = new MyWebSocketClientV2(new URI("ws://localhost:6603/websocket")); // 替换为你的WebSocket服务器地址和端点
        client.connect();
    }
}
